#include <stdbool.h>
#include "postgres.h"
#include "access/pthreadpool.h"
#include "access/xlog.h"
#include "access/xlogrecord.h"
#include <glib.h>
#include "utils/guc.h"
#include "utils/hfs.h"
GThreadPool *gpool = NULL;
static __thread GError *gerr = NULL;
static bool IsInitPool = false;
static void getWalFunc(gpointer data, gpointer user_data) {
	bool walStoreToLocal = false;
	if (EnableHotStandby && *isPromoteIsTriggered == false && !push_standby)
		walStoreToLocal = true;
	wal_batch_t* elem = (wal_batch_t*)data;
	if (elem != NULL) {
		//elem->status = STARTSTATUS;
		int r;
		//clock_t start = clock();
		r = batchRead((uint8_t *) elem->data, ThisTimeLineID2>ThisTimeLineID?ThisTimeLineID2:ThisTimeLineID, elem->startLsn, elem->endLsn, walStoreToLocal);
		//clock_t end = clock();
		//printf("====LSN %X/%X==pid %d==len %d===time %u\n",LSN_FORMAT_ARGS(elem->startLsn),pthread_self(),r,end-start);
		elem->dataLen = r;
		if (r > sizeof(XLogRecord)) {
			XLogRecord* record = ((XLogRecord*)elem->data);
			elem->startLsn = record->xl_end - record->xl_tot_len;
		} else {
			elem->startLsn = 0;
		}
		pg_atomic_exchange_u32(&elem->status,(uint32_t)COMPLETEDSTATUS);
	}
}

static void produceWalFunc(gpointer data, gpointer user_data) {
	wal_batch_t* elem = (wal_batch_t*)data;
	elem->dataLen = elem->endLsn - elem->startLsn;
	if (elem->dataLen != 0) {
		pushXlogToTikv(elem->data,elem->dataLen);
	}
	pg_atomic_exchange_u32(&elem->status,(uint32_t)COMPLETEDSTATUS);
}

int initPthreadPool(void) {
	if (IsInitPool == true) {
		return 0;
	}
	if (!g_thread_supported())
    {
        elog(FATAL,"Not support g_thread!");
        return -1;
    }
	//default 8 thread read
    if(he3mirror){
        gpool = g_thread_pool_new(produceWalFunc,NULL,4,FALSE,NULL);
    } else{
        gpool = g_thread_pool_new(getWalFunc,NULL,8,FALSE,NULL);
    }
	elog(LOG,"thread pool max threads is %d,num thread is %d",
		g_thread_pool_get_max_threads(gpool),g_thread_pool_get_num_threads(gpool));   
	return 0;
}

int WalTaskPool(wal_batch_t*data) {
	g_thread_pool_push(gpool,(gpointer)data,&gerr);
	if (gerr != NULL) {
		elog(FATAL,"WalTaskPool err %s",gerr->message);
		return -1;
	} 
	return 0;
}

void WalTaskFree(void) {
	return g_thread_pool_free(gpool,FALSE,TRUE);
}

void WalTaskImmediateFree(void) {
    g_thread_pool_free(gpool,TRUE,TRUE);
    gpool = NULL;
}

bool IsFreePthreadPool(void) {
    return gpool == NULL;
}


